import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;

@Slf4j
public class T4Routing {
    static final String EXCHANGE_NAME = "direct_logs";
    static final ExecutorService es = Executors.newFixedThreadPool(3);

    public static void main(String[] args) {
        es.submit(T4Routing::producer);
        es.submit(() -> consumer(Collections.singletonList("error")));
        es.submit(() -> consumer(Arrays.asList("error", "info", "warn")));
    }

    static void producer() {
        try (Connection connection = ConnectionUtil.getConnection();
             Channel channel = connection.createChannel()) {
            // 创建交换机，一旦创建了 exchange，RabbitMQ 是不允许对其改变的
            // 参数 1：交换机名称
            // 参数 2：交换机类型：fanout、topic、direct、headers
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
            String message = "Hello World!";
            while (true) {
                Thread.sleep(1000);
                // 发送消息
                // 参数 1：交换机名称，如果没有指定则使用默认 Default Exchage
                // 参数 2：路由 key，简单模式可以传递队列名称
                // 参数 3：消息属性：PERSISTENT_TEXT_PLAIN(将消息标记为持久性)
                // 参数 4：消息内容
                channel.basicPublish(EXCHANGE_NAME, "error", null, message.getBytes(StandardCharsets.UTF_8));
                channel.basicPublish(EXCHANGE_NAME, "info", null, message.getBytes(StandardCharsets.UTF_8));
                channel.basicPublish(EXCHANGE_NAME, "warn", null, message.getBytes(StandardCharsets.UTF_8));
                channel.basicPublish(EXCHANGE_NAME, "error", null, message.getBytes(StandardCharsets.UTF_8));
                System.out.println(Thread.currentThread().getName() + " [x] Sent '" + message + "'");
            }
        } catch (IOException | TimeoutException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void consumer(Collection<String> routingKeys) {
        try {
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
            // 创建一个非持久的、独占的、具有生成名称的自动删除队列
            String queueName = channel.queueDeclare().getQueue();
            // 队列绑定交换机
            for (String routingKey : routingKeys) {
                channel.queueBind(queueName, EXCHANGE_NAME, routingKey);
            }

            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                System.out.println(Thread.currentThread().getName() + " [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
            };
            // 订阅消息
            // 爹数 1：队列名称
            // 参数 2：是否自动确认。设置为 true 表示消息按收到后自动向 mq 回复接收到了，mq 接收到回复后会删除消息：设置为 false 则需要手动确认
            channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
            });
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
        }
    }
}